feat(sdks/go): stream reconnect and lifecycle for listeners - #4257
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
27d4227 to
2d96f31
Compare
Benchmark resultsCompared against |
|
|
|
|
|
|
b65ddd7 to
5e82a64
Compare
|
|
Rebuild stream listener lifecycle handling around shared reconnect components so action, durable, and workflow streams recover consistently without orphaned handlers or tight retry loops. Handler failures are logged and isolated instead of being treated as shared stream death, so unrelated waiters remain registered while true listener failures still surface through failAll.
5e82a64 to
6cd08fa
Compare
| // ClassifyStreamError maps a stream error to a reconnect decision. | ||
| func ClassifyStreamError(ctx context.Context, err error) StreamDecision { | ||
| if err == nil { | ||
| return StreamDecisionStop |
There was a problem hiding this comment.
this is defensive, err should be non-nill; add a comment
| continue | ||
| } | ||
|
|
||
| if rerr := s.connectSync(ctx); rerr != nil { |
There was a problem hiding this comment.
Should we honour some sort of deadline on the context so that this is not blocked for too long? The reason for the question is that I see we try StreamSyncMaxAttempts here and then StreamSyncMaxAttempts in connectSync as well which seems to stack up to a minute or so's delay unless that is going to expected behavior?
There was a problem hiding this comment.
Good catch. You are correct that the send path and the reconnect path each being "bounded at 5" compounds (up to ~25 connect attempts and roughly a minute-plus of backoff worst case). BTW this mirrors what the legacy retrySend → doRetrySubscribe on main already did, but that's not a reason to keep it.
I will flatten retrySend to make exactly one connectOnce attempt per failed send, with a single backoff per attempt.
| if !w.startListening() { | ||
| if w.isClosed() { | ||
| return errListenerClosed | ||
| remove := l.reg.store(workflowRunId, sessionId, handler, onError) |
There was a problem hiding this comment.
We store the handler before calling retrySend so if the background loop happens to reconnect in that gap, replayHandlers will already send a subscribe for this key, and then retrySend sends it again. We need to make sure for the same run ID the subscribe is idempotent on the server side it seems.
There was a problem hiding this comment.
Interesting, so I think we are fine. Here's my thinking:
- Duplicate subscribes for the same run ID on a stream are idempotent because
subscribeToWorkflowRunsV1funnels every subscribe intoworkflowRunAcks.addWorkflowRun, which is a map write (acks[id]<-false), andsendEventonly sends while the entry exists, deleting it after. - the
store->replay->retrySendrace produces an extra wire message
One caveat: delivery is at-least-once: a re-subscribe after the completion was already acked re-delivers the finished event via the poller. That behavior predates this PR, and afaict it's on purpose as handlers are duplicate-tolerant (Result() takes the first value off a buffered channel).
| } | ||
|
|
||
| if err := ctx.Err(); err != nil { | ||
| if err := l.ensureListening(lifecycle); err != nil { |
There was a problem hiding this comment.
nit: Calling this after a successful send is basically always a no-op because the gate is already active, right?
There was a problem hiding this comment.
Yeah, so after a successful send this is indeed a no-op.
It's there for the loop dying between the first ensureListening and the send completing, as without it, the handler would be registered and subscribed on the wire, but nothing would be receiving.
There's the TestWorkflowRunsListenerRestartsAfterListenExits to test for this behavior.
| case codes.Unauthenticated, codes.PermissionDenied, codes.InvalidArgument, | ||
| codes.FailedPrecondition, codes.NotFound, codes.Unimplemented: | ||
| return StreamDecisionStop | ||
| case codes.Unavailable, codes.Internal, codes.DeadlineExceeded, codes.ResourceExhausted: |
There was a problem hiding this comment.
Quick question on putting DeadlineExceeded in the retry bucket. On the sync paths the ctx.Err() checks catch a real client-side deadline, but if the server returns a DeadlineExceeded status while our context is still live, we'll keep reconnecting on it. Is that intended, or should it count as no-progress instead?
There was a problem hiding this comment.
Intended, here's my thinking:
- The classifier checks
ctx.Err()first, so a real client-side deadline stops the loop. - What lands in the retry bucket is only a DEADLINE_EXCEEDED status sent by the server and my working assumption is that on a long-lived stream that usually means a network blip cut an otherwise healthy connection --> so that's a transient failure we want to reconnect through.
- OTOH if we counted it as no-progress, a worker would "die" as a result.
WDYT?
retrySend ran a full 5-attempt connectSync round per failed send, compounding to up to 25 connect attempts and ~2 minutes of backoff on the AddWorkflowRun/AddSignal sync path. Each failed send now makes at most one coalesced connectOnce attempt before backing off, bounding the path at 5 sends, 5 reconnects, and ~15s of worst-case backoff.
|
|
|
Looking at the
I'll see if I can get an agent to harden this test in a separate PR. edit the test passed on a re-run |
2 participants
Description
Stream/listener half of #4228, building on the retry primitives from #4240 (merged).
The legacy Go SDK (
pkg/client) keeps four long-lived gRPC streams: workflow-run subscriptions, durable event subscriptions, the worker action stream, and the metadata event stream. Each had its own reconnect logic layered on top of interceptor retry, with different behaviors, e.g.Workflow.Result()could hang forever if the shared listener diedRecvfailures (which could killed the worker on a flaky network)This PR replaces that with one explicit app-level reconnect model (full-jitter backoff, 1s base / 30s cap) shared by every stream.
Behavior changes
AddWorkflowRun,AddSignal, sends): bounded reconnect — 5 attempts, then fail.Workflow.Result()fails fast when subscribing fails, and waiters are unblocked if the listener dies permanently.Unimplementedfallback preserved).Review guide
Four shared components are the main part to look at and then listeners are thin wrappers over them:
stream_classify.go: four-verdict error classifier used by both recv and reconnect paths.reconnecting_stream.go-->reconnectingStream[C]: singleflighted connects, generation counter for stale clients, bounded sync connect, lifecycle ctx owned byClose().stream_listen.go: the single flat listen loop (backoff sleeps, no-progress cap).handler_registry.go-->handlerRegistry[K,E]: RWMutex over plain maps, registration-id-guarded removal,failAllon permanent death.Type of change
Checklist
Changes have been:
Testing
go build ./... && go test ./pkg/client/... ./pkg/worker/... -count=1 -racefailAll, races).Workflow.Result()fail-fast and permanent-death notification, handler-error isolation.🤖 AI Disclosure
I acknowledge that an LLM was used in the creation of this Pull Request, in accordance with Hatchet's AI_POLICY.md.
Details: Cursor (Claude) used for implementation, test coverage, review follow-ups, and PR description drafting.